package cn.tedu.charging.order.mqtt;

import cn.tedu.charging.common.pojo.JsonResult;
import cn.tedu.charging.common.pojo.param.ChargingProcessParam;
import cn.tedu.charging.common.pojo.vo.ChargingProcessVO;
import cn.tedu.charging.common.utils.JsonUtils;
import cn.tedu.charging.order.constant.Constant;
import cn.tedu.charging.order.constant.MqttConstant;
import cn.tedu.charging.order.constant.OrderStatusConstant;
import cn.tedu.charging.order.pojo.param.ChargingResultParam;
import cn.tedu.charging.order.pojo.po.ChargingBillFailPO;
import cn.tedu.charging.order.pojo.po.ChargingBillSuccessPO;
import cn.tedu.charging.order.pojo.po.ChargingProcessPO;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.BeanUtils;

import java.math.BigDecimal;

/**
 * mqtt接收工具类，用来接收消费设备发的消息
 */
@Slf4j
//@Component    //注释原因看配置类MqttConfiguration的回调方法说明
public class MqttConsumer implements MqttCallbackExtended { //实现类，必须重写下面的4个方法

    //6.1 下面需要用mqttClient来进行发布/订阅方法，但重写的4个方法完全没涉及到mqttClient，所以考虑自动注入一个mqttClient
    //但自动注入会失败，因为MqttClient在容器里能注入，但MqttConsumer是普通类，不在容器里，不受容器控制，不能进行注入
    //所以考虑MqttConsumer加注解Component放入spring的IoC容器中
    //但要考虑让MqttConfiguration和MqttConsumer建立关联，相互知道，所以去MqttConfiguration中进行设置，在回调方法中设置。为啥这样做？？？
    //配置类MqttConfiguration的回调方法说明中，老师要new对象，不要Component注解，所以这儿注入会失败，也不要了，用构造方法。
    //但是，为啥我去掉Component注解，这个注解Autowired不报错？？？
    //因为MqttConfiguration中是new对象，所以老师说可以把构建MqttClient对象放到MqttConsumer的构造方法中，
    //配置类中new MqttConsumer时传参MqttClient，调用这个有参构造方法，从而生成mqttClient。
    //极度考验java基础语法和spring容器理解
    //@Autowired
    private MqttClient mqttClient;
    //配合mqttClient.setCallback(new MqttConsumer(mqttClient));用构造方法获取外界的mqttClient。
    /*public MqttConsumer(MqttClient mqttClient) {
        this.mqttClient = mqttClient;
    }*/

    private MqttContext mqttContext;    //7.1 给MqttConsumer的构造器多增加一个接收参数
    public MqttConsumer(MqttClient mqttClient,MqttContext mqttContext) {    //7.1 给MqttConsumer的构造器多增加一个接收参数
        this.mqttClient = mqttClient;
        this.mqttContext = mqttContext;
    }

    // 6.1 MqttConfiguration中说明，也可以用set方法获取外界的mqttClient。
    public void setMqttClient(MqttClient mqttClient) {
        this.mqttClient = mqttClient;
    }


    //重点关注连接完成、消息到达     回调方法，还是不太懂，是自动调用？
    /**
     * 连接完成,表示你已经和 EMQX消息中间件连接成功    reconnect自动重连
     * @param reconnect If true, the connection was the result of automatic reconnect.
     * @param serverURI The server URI that the connection was made to.
     */
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        log.info("MqttConsumer:连接成功");
        try {
            //和EMQX连接成功后,订阅 "/topic/start/result" 来消费消息     主题/开始/充电
            log.debug("连接成功,订阅topic:{}",MqttConstant.TOPIC_CHARGING_RESULT);
            //mqttClient.subscribe("/topic/start/result");    //发送/订阅都是mqttClient实现的,信息的发/收不建议用同一个主题，容易乱套
            mqttClient.subscribe(MqttConstant.TOPIC_CHARGING_RESULT);    //主题 改进成常量表示
            log.debug("连接成功,订阅topic:{},成功",MqttConstant.TOPIC_CHARGING_RESULT);

            log.debug("连接成功,订阅topic:{}",MqttConstant.TOPIC_CHARGING_PROCESS);
            mqttClient.subscribe(MqttConstant.TOPIC_CHARGING_PROCESS);
            log.debug("连接成功,订阅topic:{},成功",MqttConstant.TOPIC_CHARGING_PROCESS);
        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }

    @Override   //当连接丢失事件发生时，自动调用此方法
    public void connectionLost(Throwable cause) {

    }

    @Override//当从服务器接收到一条消息时，或者说服务器发来消息的事件发生时？自动调用此方法。
    //上面订阅了主题，但这儿是咋知道从哪个主题获取消息的？？咋知道上面方法的subscribe("/topic/start/result")？？？
    public void messageArrived(String topic, MqttMessage message) throws Exception {    //10.2接收桩反馈、充电成功、计价方法分开
        log.debug("MqttConsumer:消息到达:topic-{},message-{}", topic, message);
        if (MqttConstant.TOPIC_CHARGING_RESULT.equals(topic)) { //根据topic主题不同，而分库处理充电成功消息、充电进度消息
            log.debug("处理开始充电结果消息:{}",message);
            handleChargingResult(message);      //下面调用的方法都是用private修饰的
            log.debug("处理开始充电结果消息,成功:{}",message);
        }else if (MqttConstant.TOPIC_CHARGING_PROCESS.equals(topic)) {
            log.debug("处理开始充电进度消息:{}",message);
            handleChargingProcess(message);
            log.debug("处理开始充电进度消息,成功:{}",message);
        }
    }

    /**
     * 处理 设备同步 充电进度     //10.2接收桩反馈、充电成功、计价方法分开
     * @param message
     */
    private void handleChargingProcess(MqttMessage message) {
        //把设备发送的消息JSON转换为java对象ChargingProcessParam
        ChargingProcessParam chargingProcessParam = null;   //充电结果、充电进程，设置的桩反馈信息不同
        try{
            chargingProcessParam = JsonUtils.fromJson(message.toString(), ChargingProcessParam.class);
            log.debug("消息转换为ChargingProcessParam:{}",chargingProcessParam);
        }catch (Exception e) {
            log.warn("设备发送的充电进度消息格式有问题,请检查!", message);
        }
        log.debug("通过open-feign的costClient 调用计价服务 计算价格,入参:{}",chargingProcessParam);
        //1 计算电费 通过调用计价服务
        ChargingProcessVO orderCost = getOrderCost(chargingProcessParam);   //记得先在MqttContext里注入CostClient和getter
        log.debug("通过open-feign的costClient 调用计价服务 计算价格 成功.入参:{},出参:{}",chargingProcessParam,orderCost);
        if (orderCost != null) {
            BigDecimal totalCost = orderCost.getTotalCost();
            //通过调用计价服务获取当前充电进度的总花费
            log.debug("当前花费总金额:{}",totalCost);
            //2 充电进度保存  保存设备反馈的充电信息chargingProcessParam
            saveChargingProcess(chargingProcessParam);
            //3 推送充电数据给用户 推送(WebSocket)     返还给用户的数据用orderCost就行，里面有时长、费用等，所以这里传参进去
            syncChargingProcess2User(orderCost);
            //4 充电过程中如果钱不够了,如何处理?
            //  4.1 给用户发通知,钱不够了,引导用户充值 (不人性化)
            //  4.2 钱不够 继续充 直到充满为止 记录下用户欠款
            //这里是选择让桩反馈订单数据时，顺便反馈userId，反正订单成功表里有,所以chargingProcessParam里加了条userId。或者直接从订单成功表查也行。
            checkUserBalance(chargingProcessParam.getUserId(), totalCost);   //余额和总费用比较
            //5 判断温度
            //  5.1 大于 40度 小于 50度 B级告警  推送给用户 用户可以选择是否继续充
            //  5.2 大于 50度  A级告警  推送给用户 停止充电
            checkDeviceStatus(chargingProcessParam.getTemperature());   //肯定传参是温度啊
            //6 判断是否充满
            //   6.1 停止充电 更新订单状态为 订单结束
            checkChargingStatus(chargingProcessParam.getIsFull());  //给ChargingProcessParam加个属性Boolean isFull
        }
    }

    /**
     * 保存充电进度
     * 存储 存到mysql 是否可以
     * 不太行mysql性能较差  量太大 100 * 30s 一次 1分钟 2百次  1分钟产生2百条数据  平均充满电需要2个小时 就是2W4 * 12  ==
     * 数据库的每秒并发也就几百，不超过1000，所以不能直接保存到数据库。
     * ES 分布式搜索引擎
     *
     * 时序数据
     * 时间    数据
     * 10:01 充电数据
     * 10:02 充电数据
     * 10:03 充电数据
     *
     * 天气
     * 10:01 风力5  10:02 风力6   10:03 风力10
     *
     * 特点 按照时间 序列  数据量大
     * 读写特点 高频率的写 相对写 读少, 不会更新
     *
     */
    private void saveChargingProcess(ChargingProcessParam chargingProcessParam){
        log.debug("保存充电进度数据:{}",chargingProcessParam);
        //新建PO类，属性和ChargingProcessParam基本一致，除了最后的是否充满
        ChargingProcessPO chargingProcessPO = new ChargingProcessPO();
        //ES存时序数据，正好id也是有顺序的，所以把id当做时间
        chargingProcessPO.setId(System.currentTimeMillis());
        //数据
        BeanUtils.copyProperties(chargingProcessParam,chargingProcessPO);
        //12.4 保存到es
        mqttContext.getChargingProcessRepository().save(chargingProcessPO); //12.4引入ES添加
        log.debug("保存充电进度数据成功:{}",chargingProcessParam);
    }

    /**
     * 同步充电数据给用户    需要推送给手机端，用到新技术，推送(WebSocket)
     */
    private void syncChargingProcess2User(ChargingProcessVO orderCost){
        log.debug("同步充电数据到用户手机端:{}",orderCost);
        String json = JsonUtils.toJson(orderCost);  //13.1 搭建websocket
        mqttContext.getWebSocketServer().sendMessage(orderCost.getUserId(),json);
        //todo 把消息发送到RabbitMq,专门搞一个消息队列存储消息,再搞一个消费者,
        // 去消费消息,然后调用websocket(集成服务商的 比如 极光,网易云信)给用户发送消息
    }

    /**
     * 检查用户余额
     * 在用户服务，新增一个对外地址获取用户余额，就是userConstroller里加个方法，先用假数据金额替代
     * feign调用user的余额方法，所以userClient里加了余额方法。但MqttConsumer不在容器，不能直接注入userClient使用，
     * 所以在MqttContext中注入userClient传进来。这时候已经体现出来用context传参的好处了，要用的东西都封装在context中，不需要修改上面MqttConsumer的有参构造方法
     * @param userId
     * @param totalCost
     */
    private void checkUserBalance(Integer userId, BigDecimal totalCost){
        JsonResult<BigDecimal> userBalanceResult = mqttContext.getUserClient().getUserBalance(userId);  //调用用户方法
        if (userBalanceResult != null) {
            BigDecimal userBalance = userBalanceResult.getData();   //为啥要.getData()？？？
            if (userBalance.subtract(totalCost).floatValue() <= 0) {    //subtract是BigDecimal数据类型的减法
                log.debug("给用户发送通知,余额不足,停止充电.引导用户进行充值,当前余额:{},订单消费:{}",userBalance,totalCost);
                // mqttContext.getWebSocketServer().sendMessage
            }
        }
    }

    /**
     * 检查设备状态  先实现温度，后续加上电压等 todo
     */
    private void checkDeviceStatus(float temperature){
        if (temperature > MqttConstant.A_WARNING) { //加个温度常量，就走个温度监测的逻辑，A是50度，B是45度
            log.debug("温度超过:{},给用户发送A类告警,直接停止充电,实际温度:{}",MqttConstant.A_WARNING,temperature);
            //通知用户 推送消息给用户 todo  mqttContext.getWebSocketServer().sendMessage
        }else if (temperature > MqttConstant.B_WARNING) {
            log.debug("温度超过:{},给用户发送B类告警,设备不停止充电,让用户选择是否继续充电,实际温度:{}",MqttConstant.B_WARNING,temperature);
            //通知用户 推送消息给用户 todo  mqttContext.getWebSocketServer().sendMessage
        }
    }

    /**
     * 检查充电状态 判断是否已经充满
     */
    private void checkChargingStatus(Boolean isFull){
        if (isFull) {
            log.debug("已经充满,更新订单状态,更新余额");
            // mqttContext.getWebSocketServer().sendMessage
        }
    }

    /**
     * 通过open-feign的costClient 调用计价服务 计算价格  //10.2，与handleChargingProcess解耦
     * @param chargingProcessParam
     * @return
     */
    private ChargingProcessVO getOrderCost(ChargingProcessParam chargingProcessParam){//10.2接收桩反馈、充电成功、计价方法分开
        JsonResult<ChargingProcessVO> result = mqttContext.getCostClient().calculateCost(chargingProcessParam);
        if (result != null) {
            ChargingProcessVO data = result.getData();
            if (data != null) {
                return data;
            }
        }
        return null;
    }

    /**
     * 处理 设备同步 开始充电结果   //10.2接收桩反馈、充电成功、计价方法分开 直接从收到消息里，右键重构抽出
     * @param message
     */
    private void handleChargingResult(MqttMessage message) {
        //log.debug("MqttConsumer:消息到达:topic-{},message-{}", topic, message);   //
        //6.3添加，枪的反馈信息也标准化，把接收到的json格式转为java的ChargingResultParam对象
        //ChargingResultParam chargingResultParam = JsonUtils.fromJson(message.toString(), ChargingResultParam.class);
        //ChargingResultParam deviceResultParam = JsonUtils.fromJson(message.toString(), ChargingResultParam.class);
        ChargingResultParam deviceResultParam = null;   //9.3补充，有人代码格式可能有问题，所以这里catch错误
        try{
            deviceResultParam = JsonUtils.fromJson(message.toString(), ChargingResultParam.class);
            log.debug("消息转换为ChargingResultParam:{}",deviceResultParam);
        }catch (Exception e) {
            //log.warn("设备发送的消息格式有问题,请检查!",message);
            log.warn("设备发送的开始充电结果消息格式有问题,请检查!", message);
        }
        log.debug("消息到达:{}", deviceResultParam);
        //判断ChargingResultParam 不能为空
        if (deviceResultParam != null) {    //6.3补，老师才想起来飞书笔记里有个这
            //取出充电结果 1 开始成功 2 开始失败
            String result = deviceResultParam.getResult();
            //判断充电结果
            if (Constant.RESULT_START_CHARGING_SUCCESS.equals(result)) {
                //开始充电成功
                log.debug("充电桩开始充电:{}", deviceResultParam);
                //创建订单成功记录  6.4配合数据库的
                ChargingBillSuccessPO chargingBillSuccessPO = new ChargingBillSuccessPO();

                /*
                启动时，spring扫描mqtt配置类，回调方法？MqttConsumer订阅主题。
                用户传入参数，order创建订单，发送充电指令方法，mqttProducer.send方法给桩发布指令。
                回调方法MqttConsumer中在接受到枪的反馈消息后，储存订单成功/失败记录到数据库。
                所以需要考虑MqttConsumer的messageArrived中，如何保存到数据库？
                mybatis需要一个mapper并注入，然后执行sql语句保存。而和当成MqttConsumer从配置类中用构造方法获取mqttClient的原因一样，
                MqttConsumer不在spring容器中，不能外界创建一个mapper然后MqttConsumer中自动注入。
                这时可以外界建一个mapper，然后配置类MqttConfiguration中注入mapper，再在MqttConsumer中构造方法改成接收2个参数，
                然后MqttConfiguration中回调函数时的new对象时顺便传参成功表、失败表进去，但这样的话，之后每加一个参数都得改构造方法，**构造方法的参数越来越长，不好**。
                所以，老师新建一个MqttContext类，里面把需要的参数都放进去，这样构造方法就只需要传1个参数，**参数列表就短了**！！！
                然后外面新建MqttContext类，把mapper注入到MqttContext类，MqttContext类里放一堆需要的参数，
                再注入到配置类，传入到MqttConsumer构造器，再把一堆需要的参数拿出来
                 */
                chargingBillSuccessPO.setBillId(deviceResultParam.getOrderNo());    //获取桩反馈的deviceResultParam里的桩id，并写入给与数据库对应的表chargingBillSuccessPO
                chargingBillSuccessPO.setBillStatus(OrderStatusConstant.ORDER_STATUS_PROGRESS); //7.2 保存时多加一条保存数据--订单状态

                log.debug("保存成功订单数据:{}",chargingBillSuccessPO);
                //OrderSuccessMapper orderSuccessMapper = mqttContext.getOrderSuccessMapper();  //能看到得到的就是个OrderSuccessMapper了
                //mqttContext.getOrderSuccessMapper().insert(chargingBillSuccessPO);  //mqttContext中得到OrderSuccessMapper，执行mybatis-plus的insert语句插入
//                mqttContext.getOrderRepository().saveSuccessOrder(chargingBillSuccessPO);//8.5 数据层解耦  9.4下面catch异常
                log.debug("保存成功订单数据:{},成功",chargingBillSuccessPO);

                //9.4git 就是给数据库多加了唯一索引订单号，
                // 捕获异常就是为了这时候查看一下订单号是唯一索引时，重复消息的数据库的报错。订单号是唯一索引时，不捕获的话也没影响，直接就不执行了，就是看不到内部原理
                //如果设备重复发送消息，会导致 成功订单表 charging_bill_success 会重复的保存订单数据
                //如何解决
                //1 给订单表 订单编号 创建一个唯一索引,唯一索引的意思 这个表里不能有重复的订单编号
                // 加完唯一索引后,如果有重复的记录要保存 数据库会抛出异常
                // Duplicate entry '10000_58_1721720859912' for key 'charging_bill_success_bill_id_unique_key'
                // 从而避免订单成功表有重复的订单记录
                // 异常可以处理,可以不处理————通常接入监控系统，如果固定的时间之内，重复发送消息,设备出问题了,通知调用方 设备维修人员
                //2 先查询订单成功表是否有数据,如果没有 插入,如果有 跳过不保存, 代码怎么写? todo 原子性的问题 但高并发时，很难保证原子性？
                //3 记录状态 如果处理过了,不再重复处理  redis的布隆过滤器 自己了解 todo
                try{
                    mqttContext.getOrderRepository().saveSuccessOrder(chargingBillSuccessPO);
                }catch (Exception e) {
                    log.error("保存成功订单数据失败",e);
                }

            } else {
                log.debug("充电状态开始充电失败,充电桩故障.");
                //创建订单失败记录  6.4配合数据库的
                //orderService.saveFailOrder();
                ChargingBillFailPO chargingBillFailPO = new ChargingBillFailPO();
                chargingBillFailPO.setBillId(deviceResultParam.getOrderNo());
                //后续处理无响应的设备 todo
                chargingBillFailPO.setFailDesc("设备自检失败-设备有响应但是充不了电");
                log.debug("保存失败订单数据:{}",chargingBillFailPO);
                //mqttContext.getOrderFailMapper().insert(chargingBillFailPO);//8.5 数据层解耦
                mqttContext.getOrderRepository().saveFailOrder(chargingBillFailPO);
                log.debug("保存失败订单数据:{},成功",chargingBillFailPO);
            }
        }

    }

    @Override   //当一个消息被确认送达时，调用此方法
    //delivery交付
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.debug("MqttConsumer deliveryComplete");
    }


}
